package drds.data_migrate.job.read_and_write;

import com.google.common.collect.Lists;
import drds.data_migrate.util.ReaderSplitUtil;
import drds.data_migrate.util.Reads;
import drds.data_migrate.util.TableCheckTask;
import drds.data_migrate.vo.ConnectionInfo;
import drds.data_migrate.vo.JobConfig;
import drds.data_migrate.vo.TaskConfig;
import lombok.Getter;
import lombok.Setter;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * Created by ibm on 2018/11/3.
 */
public class ReadJob
{

    @Setter
    @Getter
    private JobConfig jobConfig;

    public ReadJob()
    {


    }

    public ReadJob(JobConfig jobConfig)
    {

        this.jobConfig = jobConfig;
    }

    public void init() throws SQLException
    {

        Reads.doPretreatment(jobConfig);

    }

    public void preCheck()
    {
        /*检查每个表是否有读权限，以及querySql跟splik Key是否正确*/
        ReaderSplitUtil.doPreCheckSplit(jobConfig);


        List<ConnectionInfo> connectionInfoList = jobConfig.getConnectionInfoList();

        //
        ExecutorService executorService;
        if (connectionInfoList.size() < 10)
        {
            executorService = Executors.newFixedThreadPool(connectionInfoList.size());
        } else
        {
            executorService = Executors.newFixedThreadPool(10);
        }
        Collection<TableCheckTask> tableCheckTaskList = new ArrayList<TableCheckTask>();
        for (int i = 0, size = connectionInfoList.size(); i < size; i++)
        {
            ConnectionInfo connectionInfo = connectionInfoList.get(i);
            TableCheckTask tableCheckTask = new TableCheckTask(connectionInfo);
            tableCheckTaskList.add(tableCheckTask);
        }
        List<Future<Boolean>> futureList = Lists.newArrayList();
        try
        {
            futureList = executorService.invokeAll(tableCheckTaskList);
        } catch (InterruptedException e)
        {
            Thread.currentThread().interrupt();
        }

        for (Future<Boolean> future : futureList)
        {
            try
            {
                boolean ok = future.get();
                if (!ok)
                {
                    throw new IllegalStateException();
                }
            } catch (ExecutionException e)
            {
                throw new RuntimeException(e);
            } catch (InterruptedException e)
            {
                Thread.currentThread().interrupt();
            }
        }
        executorService.shutdownNow();
    }

    /**
     * 切分任务
     *
     * @param adviceNumber 着重说明下，adviceNumber是框架建议插件切分的任务数，插件开发人员最好切分出来的任务数>=
     *                     adviceNumber。<br>
     *                     <br>
     *                     之所以采取这个建议是为了给用户最好的实现，例如框架根据计算认为用户数据存储可以支持100个并发连接，
     *                     并且用户认为需要100个并发。 此时，插件开发人员如果能够根据上述切分规则进行切分并做到>=100连接信息，
     *                     DataX就可以同时启动100个Channel，这样给用户最好的吞吐量 <br>
     *                     例如用户同步一张Mysql单表，但是认为可以到10并发吞吐量，插件开发人员最好对该表进行切分，比如使用主键范围切分，
     *                     并且如果最终切分任务数到>=10，我们就可以提供给用户最大的吞吐量。 <br>
     *                     <br>
     *                     当然，我们这里只是提供一个建议值，Reader插件可以按照自己规则切分。但是我们更建议按照框架提供的建议值来切分。 <br>
     *                     <br>
     *                     对于ODPS写入OTS而言，如果存在预排序预切分问题，这样就可能只能按照分区信息切分，无法更细粒度切分，
     *                     这类情况只能按照源头物理信息切分规则切分。 <br>
     *                     <br>
     */
    public List<TaskConfig> split(
            int adviceNumber) throws SQLException
    {
        return ReaderSplitUtil.doSplit(jobConfig, adviceNumber);
    }

    public void post()
    {
        // do nothing
    }

    public void destroy()
    {
        // do nothing
    }

}
